package com.atguigu.gmall2020.realtime.spark.util

import java.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.Jedis


/**
 * @deprecated:
 * @author: tiancy
 * @create: 2021/11/3
 */
object OffsetManager {

  /**
   * @deprecated : 当程序挂了后,从Redis中读取最后提交的偏移量,作为数据源的初始偏移量
   * @param topic   :参数1 传入一个主题
   * @param groupId : 参数2 传入一个消费者组
   * @return : offsetMapForKafka 返回一个指定格式的offset,这个格式是Kafka中一个指定类格式.Map[TopicPartition, Long]
   */
  def getOffset(topic: String, groupId: String): Map[TopicPartition, Long] = {
    val jedis: Jedis = RedisUtil.getJedisClient
    //redis  type ?  hash    key ?  offset:[topicName]:[groupId]      field ? partition_id  value?  offset
    // 写入api ？  hset hmset   读取api？ hgetall    超时时间？ 不设过期 永久保留

    val offsetKey = "offset:" + topic + ":" + groupId
    //从redis中取值
    val offsetMap: util.Map[String, String] = jedis.hgetAll(offsetKey)
    //转换格式,通过隐式转换将Java的Map类型转化为Scala中的Map.再使用map方法.
    import scala.collection.JavaConverters._
    val offsetMapForKafka: Map[TopicPartition, Long] = offsetMap.asScala.map {
      case (partitionId, offset) =>
        val topicPartition = new TopicPartition(topic, partitionId.toInt)
        (topicPartition, offset.toLong)
    }.toMap
    println("加载redis中的偏移量：" + offsetMapForKafka)

    jedis.close
    offsetMapForKafka
  }


  /**
   * @deprecated :  偏移量的写入到Redis的方法
   * @param topic        : 提交偏移量的主题名称
   * @param groupId      : 提交偏移量的消费者组
   * @param offsetRanges : 当前消费者组提交的最终的偏移量
   */
  def saveOffset(topic: String, groupId: String, offsetRanges: Array[OffsetRange]): Unit = {
    val jedis: Jedis = RedisUtil.getJedisClient
    // 把offsetRanges 整理成map[String,String]
    val redisOffsetMap = new util.HashMap[String, String]()

    for (offsetRange <- offsetRanges) {
      val partitionId: Int = offsetRange.partition // 获取到的分区数
      val offset: Long = offsetRange.untilOffset //获取到的本批次最终的偏移量
      redisOffsetMap.put(partitionId.toString, offset.toString)
    }

    val offsetKey = "offset:" + topic + ":" + groupId
    /*
      TODO 往Redis中存入执行格式的偏移量.Map[分区号,当前分区号中的偏移量]
     */
    jedis.hset(offsetKey, redisOffsetMap)
    println("向redis写入偏移量：" + redisOffsetMap)
    jedis.close()
  }
}